package com.lagou.streamsink;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.io.Serializable;
import java.sql.*;


/**
 * <dependency>
 * <groupId>com.github.housepower</groupId>
 * <artifactId>clickhouse-native-jdbc</artifactId>
 * <version>1.6-stable</version>
 * </dependency>
 */
public class MySinkToClickHouse {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        ChBean b1 = new ChBean(System.currentTimeMillis(), "lucas", 20);
        ChBean b2 = new ChBean(System.currentTimeMillis(), "lucas", 21);
        ChBean b3 = new ChBean(System.currentTimeMillis(), "lucas", 22);
        ChBean b4 = new ChBean(System.currentTimeMillis(), "lucas", 23);
        ChBean b5 = new ChBean(System.currentTimeMillis(), "lucas", 24);

        DataStreamSource<ChBean> data = env.fromElements(b1, b2, b3, b4, b5);

        String chDriver = "com.github.housepower.jdbc.ClickHouseDriver";
        Class.forName(chDriver);
        String chUrl = "jdbc:clickhouse://hadoop4:9002";
//        String chUrl = "jdbc:clickhouse://hdp-1:9000";
        data.addSink(new RichSinkFunction<ChBean>() {
            Connection conn;
            PreparedStatement pst;


            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("===================open");
                conn = DriverManager.getConnection(chUrl);

//                statement.executeQuery("create table default.jdbc_example(day Date, name String, age UInt8) Engine=Log");

            }

            @Override
            public void invoke(ChBean value, Context context) throws Exception {
                System.out.println("value:" + value);
                String sql = "insert into default.lucas (day,name,age) values (?,?,?)";
                pst = conn.prepareStatement(sql);
                pst.setDate(1, new Date(value.getDate()));
                pst.setString(2, value.getName());
                pst.setInt(3, value.getAge());

                pst.execute();
            }

            @Override
            public void close() throws Exception {
                if (conn != null) {
                    conn.close();
                }
                if (pst != null) {
                    pst.close();
                }
            }
        });

        env.execute();

    }
}

class ChBean implements java.io.Serializable {
    private long date;
    private String name;
    private int age;

    public ChBean(long date, String name, int age) {
        this.date = date;
        this.name = name;
        this.age = age;
    }

    public long getDate() {
        return date;
    }

    public void setDate(long date) {
        this.date = date;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    @Override
    public String toString() {
        return "ChBean{" +
                "date=" + date +
                ", name='" + name + '\'' +
                ", age=" + age +
                '}';
    }
}